package producersConsumersTests.utils;

import java.util.Random;

import prod_cons.TaskContainer;

public class Inserter<TASK extends RunnableWithId> implements Runnable {
	protected TaskContainer container;
	protected int peakLength; // num of tasks to insert during the peak
	// milliseconds to wait between peaks (in the average)
	protected int timeBetweenPeaks;
	protected volatile boolean stop = false;
	protected TaskFactory<TASK> taskFactory;
	protected long taskId = 0;
	protected Random random = new Random();
	protected double lambda;

	public enum ArrivalType {
		POISSON, BURSTS
	};

	private final ArrivalType arrivals;

	public Inserter(TaskContainer container, int peakLength,
			int timeBetweenPeaks, TaskFactory<TASK> taskFactory) {
		this.container = container;
		this.peakLength = peakLength;
		this.timeBetweenPeaks = timeBetweenPeaks;
		this.taskFactory = taskFactory;
		this.arrivals = ArrivalType.BURSTS;
	}

	public Inserter(TaskContainer container, double lambda,
			TaskFactory<TASK> taskFactory) {
		this.container = container;
		this.lambda = lambda;
		this.taskFactory = taskFactory;
		this.arrivals = ArrivalType.POISSON;
	}

	public final void stop() {
		this.stop = true;
	}

	protected void beforeRun() {
	}

	protected void afterRun() {
	}

	protected void beforeInsert(TASK task) {
	}

	protected void afterInsert(TASK task) {
	}

	protected void afterBurst() {
	}

	protected void beforeBurst() {
	}

	@Override
	public final void run() {
		if (arrivals == ArrivalType.BURSTS) {
			burstsRun();
		} else {
			poissonRun();
		}
	}

	private final void burstsRun() {
		long timeToSleep = 0;
		beforeRun();
		while (!stop) {
			beforeBurst();
			// peak period
			for (int i = 0; i < this.peakLength; i++) {
				TASK task = taskFactory.createTask(taskId++);
				beforeInsert(task);
				this.container.put(task);
				afterInsert(task);
			}
			afterBurst();
			// silent period
			try {
				timeToSleep = Math.round(timeBetweenPeaks
						+ Math.sqrt(timeBetweenPeaks) * random.nextGaussian());
				if (timeToSleep > 0)
					Thread.sleep(timeToSleep);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		afterRun();
	}

	private void poissonRun() {
		beforeRun();
		while (!stop) {
			try {
				long timeToSleep = generatePoissonNumber(lambda);
				if (timeToSleep > 0)
					Thread.sleep(timeToSleep);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
			TASK task = taskFactory.createTask(taskId++);
			beforeInsert(task);
			this.container.put(task);
			afterInsert(task);
		}
	}

	private ThreadLocal<Random> rand = new ThreadLocal<Random>() {
		@Override
		protected Random initialValue() {
			return new Random(1);
		}
	};

	protected final long generatePoissonNumber(double lambda) {
		double l = Math.exp(-lambda), p = 1;
		long k = 0;
		while (p > l) {
			k++;
			p *= rand.get().nextDouble();
		}
		return k;
	}
}
